#include <ctype.h>
#include <stdint.h>
#include <sys/eventfd.h>

#include "game_thread.h"
#include "game_event.h"
#include "logic_thread.h"
#include "msg_queue.h"
#include "kafka_thread.h"

struct thread_logic_param
{
	struct normal_queue nq;
	pthread_cond_t       cond;
	pthread_mutex_t      mutex;
	logic_msg_callback callback;
};
static struct thread_logic_param logic_thread_param;

static void run_logic_callback(struct thread_logic_param *param, struct message_queue *q)	
{
	while (q)
	{
		struct send_msg *msg = (struct send_msg *)q->data;
		UNUSED(msg);

		logic_thread_param.callback(q->data);
		q = q->next;
	}
}

void *thread_logic(void *p)
{
	struct thread_logic_param *param = &logic_thread_param;

	while (g_thread_run) {
		struct message_queue *q = normal_queue_getmq(&param->nq);
		if (q)
		{
			run_logic_callback(param, q);
		}
		else
		{
			if (pthread_mutex_lock(&param->mutex) == 0)
			{
				if (g_thread_run)
					pthread_cond_wait(&param->cond, &param->mutex);
				if (pthread_mutex_unlock(&param->mutex))
				{
					LOG_ERR( "unlock mutex error");
					exit(1);
				}
			}
		}
	}

	return NULL;
}

int init_thread_logic_param(logic_msg_callback callback)
{
	normal_queue_init(&(logic_thread_param.nq));	

	if (pthread_mutex_init(&logic_thread_param.mutex, NULL)) {
		fprintf(stderr, "Init mutex error");
		exit(1);
	}
	if (pthread_cond_init(&logic_thread_param.cond, NULL)) {
		fprintf(stderr, "Init cond error");
		exit(1);
	}
	logic_thread_param.callback = callback;
	return (0);
}
pthread_t create_kafka_logic_thread(logic_msg_callback callback, void *param)
{
	pthread_t pid;
	init_thread_logic_param(callback);
	create_thread(&pid, thread_logic, param);
	return pid;
}


int wakeup_logic_thread(int force)
{
	if (force > 0 || !is_normal_queue_empty(&logic_thread_param.nq))
	{
		pthread_cond_signal(&logic_thread_param.cond);
	}
	return (0);
}

int send_to_logic(void *data, int len)
{
	// struct send_msg *msg_data = (struct send_msg *)malloc(sizeof(struct send_msg));
	struct message_queue *msg = (struct message_queue *)malloc(sizeof(struct message_queue));
	if (!msg)
	{
		fprintf(stderr,
		        "%% mallooc error: %d\n",
		        errno);
		return -1;
	}
	msg->cmd  = 0;
	msg->len  = len;
	msg->data = data;
	msg->next = NULL;
	normal_queue_push(&logic_thread_param.nq, msg);
	return (0);
}
